package com.ehualu.liaocheng

import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * @author ：吴敬超
  */
object realTimekafka {







  def main(args: Array[String]): Unit = {


    security.security.run()

    //KAFKA安全认证
    //    security.run()
    //创建SparkConf，如果将任务提交到集群中
    val conf = new SparkConf().setAppName("realTimeMonitoringtest").setMaster("local[2]")
    //创建一个StreamingContext，批次时长为2秒
    val ssc = new StreamingContext(conf, Seconds(2))


    ssc.sparkContext.setLogLevel("WARN")
    //配置工厂 获取kakfa配置信息
    val load = ConfigFactory.load("kafka")

    //消费卡夫卡参数配置
    val kafkaParams = Map[String, Object](
      ///brokers 地址
      "bootstrap.servers" -> load.getString("kafka.brokers"),
      //指定该 consumer 将加入的消费组
      "group.id" -> load.getString("kafka.groupid"),
      //指定序列化类
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      // 是否开启自动提交 offset
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    //主题列表
    val topics = Array(load.getString("kafka.topics"))

    //直连方式消费
    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,
      PreferConsistent,
      //订阅的策略（可以指定用正则的方式读取topic，比如my-ordsers-.*）
      Subscribe[String, String](topics, kafkaParams))


    val data = stream.map(

      line => {


        line.value()
      }

    )


    //    data.print()


//        val dataRDD = data.reduceByWindow(_, Seconds(10), Seconds(2))


//    val dataRDD = data.countByWindow(Seconds(10), Seconds(2))


//    println(dataRDD)


    //reducefunction计算每个rdd的和，60s是窗口，20是滑动步长
    //    val hottestDStream = data.reduceByKeyAndWindow((v1:Int,v2:Int) => v1 + v2, Seconds(60) ,Seconds(20))


    //    val streamval = stream.foreachRDD(rdd => {
    //
    //      rdd.map(
    //        mes => {
    //
    //          println("************88")
    //          println(mes.value())
    //        }
    //
    //      )
    //
    //    })


    //遍历批次
    //    stream.foreachRDD(rdd => {
    //
    //
    //      rdd.foreachPartition(t => {
    //
    //        t.foreach(println)
    //
    //
    //      })
    //
    //
    //    })

    //启动
    ssc.start()
    //等待退出
    ssc.awaitTermination()

  }
}
